/*
 * DbConnection.cpp
 *
 *  Created on: 2018年1月3日
 *      Author: zhengboyuan
 */

#include "DbConnection.h"
#include "DbUtil.h"
#include "DbConnectionPool.h"
#include "DbPoolLock.h"
#include "CLog.h"
#include "SociType.h"



namespace db
{

static const int TRY_UPDATE_COUNT = 2;
static const int TRY_QUERY_COUNT = 2;
static const int TRY_UPDATE_INTERVAL = 50; //ms


DbConnection::DbConnection(DbConnectionPool* pool):
		m_pool(pool),
		m_owner(new soci::session(pool->getPool())),
        m_session(m_owner)
{
    if (!pool->isOpen())
    {
        pool->reconnect();
    }
}

DbConnection::DbConnection(DbConnectionPool& pool):
        m_pool(&pool),
        m_owner(new soci::session(pool.getPool())),
        m_session(m_owner)
{
    if (!pool.isOpen())
    {
        pool.reconnect();
    }
}

DbConnection::DbConnection(soci::session& session):
        m_pool(),
		m_owner(),
        m_session(&session)
{
}



DbConnection::~DbConnection()
{
    if (m_owner)
    {
        delete m_owner;
    }
}

bool DbConnection::validate()
{
	int category = soci::soci_error::unknown;
	std::string err;

    bool got = DbUtil::validate(*m_session, getSourceType(), category, err);
	if (!got)
	{
		if (DbUtil::isCrashed(getSourceType(), category, err) || true)
		{
			CLog::warning("db connection is crashed.\n");

			m_pool->reopen();

			reopen();
			got = DbUtil::validate(*m_session, getSourceType(), category, err);

			//got = false;
		}
	}
	return got;
}

bool DbConnection::isCrashed()
{
	int category = soci::soci_error::unknown;
	std::string err;

	bool got = DbUtil::validate(*m_session, getSourceType(), category, err);
	if (!got)
	{
		return DbUtil::isCrashed(getSourceType(), category, err);
	}
	return false;
}

bool DbConnection::doInsert(const char* sql, const char* table, int64_t& id, bool& gotId)
{
    DbPoolLock lock(m_pool);

    if (!doExecute(sql))
    {
        return false;
    }

    std::string t(table);
    gotId = m_session->get_last_insert_id(t, (long&)id);
    return true;
}

bool DbConnection::insert(const char* sql, const char* table, int64_t& id)
{
    bool gotId = false;
    bool done = doInsert(sql, table, id, gotId);
    if (done && !gotId)
    {
        if (isSqlite())
        {
            gotId = queryValue("select last_insert_rowid()", id);
        }
    }
	return done;
}

int DbConnection::update(const char* sql)
{
    DbPoolLock lock(m_pool);

    int i = 0;
    for (i = 0; i < TRY_UPDATE_COUNT; i ++)
    {
        try
        {
            soci::statement st(*m_session);
            st.alloc();
            st.prepare(sql);
            st.define_and_bind();
            st.execute(true);

            return (int)st.get_affected_rows();
        }
        catch(soci::soci_error& ex)
        {
			soci::soci_error::error_category category = ex.get_error_category();

            if (isSqlite())
            {
                CLog::error("DbConnection::update. soci error: %s\n", ex.what());
                DbUtil::sleepMs(TRY_UPDATE_INTERVAL);
                continue;
            }

			if (category == soci::soci_error::connection_error || category == soci::soci_error::unknown)
			{
				if (reconnect())
				{
					continue;
				}
			}

            CLog::error("DbConnection::update. category: %d, soci error: %s\n", category, ex.what());
            return 0;
        }
        catch (...)
        {
            CLog::error("DbConnection::update. error\n");
            return 0;
        }
    }

    if (i >= TRY_UPDATE_COUNT)
    {
        CLog::error("DbConnection::update. error. try count exceed TRY_UPDATE_COUNT. sql: %s\n", sql);
    }

    return 0;
}


bool DbConnection::doExecute(const char* sql)
{
    int i = 0;
    for (; i < TRY_UPDATE_COUNT; i++)
    {
        try
        {
            soci::statement st(*m_session);
            st.alloc();
            st.prepare(sql);
            st.define_and_bind();
            st.execute(true);
            return true;
        }
        catch (soci::soci_error& ex)
        {
			soci::soci_error::error_category category = ex.get_error_category();

			if (isSqlite())
			{
				CLog::error("DbConnection::execute. soci error: %s\n", ex.what());
				DbUtil::sleepMs(TRY_UPDATE_INTERVAL);
				continue;
			}
			
			if (category == soci::soci_error::connection_error || category == soci::soci_error::unknown)
			{
				if (reconnect())
				{
					continue;
				}
			}

            CLog::error("DbConnection::execute. category: %d, soci error: %s\n", category, ex.what());
            return false;
        }
        catch (...)
        {
            CLog::error("DbConnection::execute. error\n");
            return false;
        }
    }

    if (i >= TRY_UPDATE_COUNT)
    {
        CLog::error("DbConnection::execute. error. try count exceed TRY_UPDATE_COUNT. sql: %s\n", sql);
    }

    return false;
}

bool DbConnection::execute(const char* sql)
{
    DbPoolLock lock(m_pool);
    return doExecute(sql);
}

bool DbConnection::query(const char* sql, RowMapper mapper)
{
    DbPoolLock lock(m_pool);

	for (int i = 0; i < TRY_QUERY_COUNT; i++)
	{
		try
		{
			soci::row row;
			soci::statement st(*m_session);
			st.exchange(soci::into(row));
			st.alloc();
			st.prepare(sql);
			st.define_and_bind();
			st.execute(true);

			bool got = m_session->got_data();
			if (got)
			{
				DbRow r(row);
				got = mapper(r);
			}
			return got;
		}
		catch (soci::soci_error& ex)
		{
			soci::soci_error::error_category category = ex.get_error_category();

            CLog::error("DbConnection::query. category: %d, soci error: %s\n", category, ex.what());

			if (category == soci::soci_error::connection_error || category == soci::soci_error::unknown)
			{
				if (reconnect())
				{
					if (isSqlite())
					{
						DbUtil::sleepMs(TRY_UPDATE_INTERVAL);
					}

					continue;
				}
			}
			
			return false;
		}
		catch (...)
		{
			return false;
		}
	}
    return false;
}


static std::auto_ptr< soci::rowset<soci::row> > queryRow(DbConnection& conn, soci::session& session, const std::string& sql)
{

    std::auto_ptr< soci::rowset<soci::row> > rowset;
    for (int i = 0; i < TRY_QUERY_COUNT; i ++)
    {
        try
        {
            //soci::details::prepare_temp_type temp = (m_session->prepare << sql);
            //rowset.reset(new soci::rowset<soci::row>(temp));
            //temp.get_prepare_info()->dec_ref();

            soci::rowset<soci::row> rs = (session.prepare << sql);
            rowset.reset(new soci::rowset<soci::row>(rs));
        }
        catch(soci::soci_error& ex)
        {
			soci::soci_error::error_category category = ex.get_error_category();

            CLog::error("DbConnection::query rowser. category: %d, soci error: %s\n", category, ex.what());

			if (category == soci::soci_error::connection_error || category == soci::soci_error::unknown)
			{
				if (conn.reconnect())
				{
					if (conn.isSqlite())
					{
						DbUtil::sleepMs(TRY_UPDATE_INTERVAL);
					}

					continue;
				}
			}
        }
    }
    return rowset;
}

int DbConnection::queryRows(const char* sql, RowMapper mapper)
{
    DbPoolLock lock(m_pool);

	std::auto_ptr< soci::rowset<soci::row> > rowset = queryRow(*this, *m_session, sql);
	if (!rowset.get())
	{
		return 0;
	}

	int count = 0;
	soci::rowset<soci::row>::iterator it = rowset->begin();
	for (; it != rowset->end(); ++it)
	{
		DbRow r(*it);

		try
		{
			if (mapper(r))
			{
				count ++;
			}
			else
			{
			    break;
			}
		}
		catch (...)
		{
			break;
		}
	}
	return count;
}


bool DbConnection::reconnect()
{
    try
    {
        m_session->reconnect();
        return true;
    }
    catch (soci::soci_error& ex)
    {
        CLog::error("DbConnection::reconnect. soci error: %s\n", ex.what());
        return false;
    }
}

soci::session& DbConnection::getSession()
{
	return *m_session;
}

DbSourceType DbConnection::getSourceType()
{
	if (m_pool)
	{
		return m_pool->getSource().type;
	}

	DbSourceType t = DB_NONE;
	std::string name = DbUtil::getBackendName(*m_session);
	DbSource::parse(name.c_str(), t);
	return t;
}

bool DbConnection::isSqlite()
{
	return getSourceType() == DB_SQLITE;
}

bool DbConnection::isMysql()
{
    return getSourceType() == DB_MYSQL;
}

bool DbConnection::queryTime(const char* sql, time_t& value)
{
    RowMapper mapper = [&] (DbRow& r)
    {
        r.getTime((size_t)0, value);
        return true;
    };
    return query(sql, mapper);
}

void DbConnection::reopen()
{
	if (m_owner)
	{
		delete m_owner;
		m_owner = nullptr;
	}

	m_owner = new soci::session(m_pool->getPool()),
	m_session = m_owner;
}




} /* namespace db */
